package com.mjf.streaming

import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time

/**
 * 流处理
 */
object WordCount {
  def main(args: Array[String]): Unit = {

    // 创建 Flink 上下文环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置运行模式为 流处理
    env.setRuntimeMode(RuntimeExecutionMode.STREAMING)
    // 设置并行度为1
    env.setParallelism(1)

    // source(从 socket 读取数据)
    // 程序启动前需要使用 `nc -lk 9999` 开启一个端口
    val source: DataStream[String] = env.socketTextStream("hadoop103", 9999)

    // transformation
    val transformationResult: DataStream[(String, Int)] = source
      .flatMap(_.split("\\W+"))
      .map(word => (word.toLowerCase, 1))
      .keyBy(_._1)
      .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
      .sum(1)

    // sink
    transformationResult.print()

    // 执行job
    env.execute(WordCount.getClass.getName)

  }
}
